package hbase

/**
 * @author Hanhan_Jun
 * @date 2022年03月08日 14:26
 */
import org.apache.spark.sql.SparkSession
import org.apache.spark.{SparkConf, SparkContext}
//import com.redislabs.provider.redis._
import org.apache.spark.rdd.RDD
import redis.clients.jedis.{JedisPool, JedisPoolConfig}
import redis.clients.jedis.{Jedis, JedisPool, Response,Pipeline}

object SparkProcess {
  def main(args: Array[String]): Unit = {

    //连接spark框架
    val spark = SparkSession.builder()
      .master("local[2]")
      .appName("DemandAnalysis")
      .getOrCreate()

    //设置日志打印级别
    spark.sparkContext.setLogLevel("WARN")
    val sc = spark.sparkContext

    //读取信息表
    val Station = spark.sparkContext.textFile("C:\\Users\\karri\\Desktop\\大三下\\大数据存储课设\\all\\all.csv")
      .zipWithIndex().filter(_._2>=1).map(x=>x._1)
    val holidayRDD = spark.sparkContext.textFile("C:\\Users\\karri\\Desktop\\大三下\\大数据存储课设\\holiday.csv")
      .zipWithIndex().filter(_._2>=1).map(x=>x._1)

    //redis连接
    val redisHost = "127.0.0.1"
    val redisPort = 6379
    val redisClient = new Jedis(redisHost, redisPort)
    //val redisPassword = "123456"
    //redisClient.auth(redisPassword)
    redisClient.select(1)
    redisClient.flushAll()

    //1.总客流量----数字
    val num = Station.count()

    redisClient.set("num", num.toString)

    //2.平均消费额（）--------数字
    val ave_money = Station.filter(line => (line.trim().length > 0) && (line.split(",").length == 17))
      .map(x => (1, x.split(",")(16).toFloat))
      .reduceByKey(_ + _)
      .map(x => {
        val ave = x._2.toFloat / num.toFloat
        (1.0, ave)
      }).collect()
      .foreach(x => redisClient.set("ave_money", s"${x._2}"))
    //3.线路客流量排行（1号线>2号线....）-----柱状图
    val enter_line = Station.filter(line => (line.trim().length > 0) && (line.split(",").length == 17))
      .map(x => (x.split(",")(3), 1))
      .reduceByKey(_ + _)
    val out_line = Station.filter(line => (line.trim().length > 0) && (line.split(",").length == 17))
      .map(x => (x.split(",")(7), 1))
      .reduceByKey(_ + _)
    val all_line = enter_line.join(out_line).map(x => (x._1, (x._2._1 + x._2._2)))
      .sortBy(_._2, false)
      .collect()
      .foreach(x => redisClient.hset("all_line", s"${x._1}", s"${x._2}"))
    //4.各站点客流量排行（各站点）-------排行榜
    val enter_station = Station.filter(line => (line.trim().length > 0) && (line.split(",").length == 17))
      .map(x => (x.split(",")(2), 1))
      .reduceByKey(_ + _)
    val out_station = Station.filter(line => (line.trim().length > 0) && (line.split(",").length == 17))
      .map(x => (x.split(",")(6), 1))
      .reduceByKey(_ + _)
    val all_station = enter_station.join(out_station).map(x => (x._1, (x._2._1 + x._2._2)))
      .sortBy(_._2, false)
      .collect()
      .foreach(x => redisClient.hset("all_station", s"${x._1}", s"${x._2}"))
    //5.各站点入站客流量排行---------排行榜
    val enter_station1 = Station.filter(line => (line.trim().length > 0) && (line.split(",").length == 17))
      .map(x => (x.split(",")(2), 1))
      .reduceByKey(_ + _)
      .sortBy(_._2, false)
      .collect()
      .foreach(x => redisClient.hset("enter_station", s"${x._1}", s"${x._2}"))
    //6.各站点出站客流量排行---------排行榜
    val out_station1 = Station.filter(line => (line.trim().length > 0) && (line.split(",").length == 17))
      .map(x => (x.split(",")(6), 1))
      .reduceByKey(_ + _)
      .sortBy(_._2, false)
      .collect()
      .foreach(x => redisClient.hset("out_station", s"${x._1}", s"${x._2}"))
    //以上三个可以做成一副图
    //7.不同时间段（小时）总客流量排行（只统计入站人数）--------折线图
    val time_hour = Station.filter(line => (line.trim().length > 0) && (line.split(",").length == 17)) //去除空行和属性数不为17的数据
      .map(x => (x.split(",")(5).split(":")(0), 1))
      .reduceByKey(_ + _)
      .sortBy(_._2, false)
      .collect()
      .foreach(x => redisClient.hset("time_hour", s"${x._1}", s"${x._2}"))
    //8.不同时间段(每月)的客流量排名---------折线图
    val time_month = Station.filter(line => (line.trim().length > 0) && (line.split(",").length == 17)) //去除空行和属性数不为17的数据
      .map(x => (x.split(",")(4).split("-")(1), 1))
      .reduceByKey(_ + _)
      .sortBy(_._2, false)
      .collect()
      .foreach(x => redisClient.hset("time_month", s"${x._1}", s"${x._2}"))
    //以上两个可以做出一幅图
    //9.不同线路的客流组成（占比）------四个饼图
    val all_line2 = enter_line.join(out_line).map(x => (x._1, (x._2._1 + x._2._2)))
    val lines_all = all_line2.collect().toList
    var line_1_all = 0
    var line_2_all = 0
    var line_3_all = 0
    var line_4_all = 0

    for (line <- lines_all) {
      if (line._1 == "1") {
        line_1_all = line._2

      } else if (line._1 == "2") {
        line_2_all = line._2
      } else if (line._1 == "3") {
        line_3_all = line._2
      } else if (line._1 == "4") {
        line_4_all = line._2
      }
    }

    val lines = Station.filter(line => (line.trim().length > 0) && (line.split(",").length == 17))
    val line_diff =
      Station.filter(line => (line.trim().length > 0) && (line.split(",").length == 17)
        && (line.split(",")(3) != line.split(",")(7)))

    //1号线
    val line_1_enter = lines.filter(line => (line.split(",")(3) == "1")).map(x => (x.split(',')(2), 1)).reduceByKey(_ + _)
    val line_1_out = line_diff.filter(line => (line.split(",")(7) == "1")).map(x => (x.split(',')(6), 1)).reduceByKey(_ + _)
    val line_1 = line_1_enter.join(line_1_out)
      .map(x => (x._1, ((x._2._1 + x._2._2) / line_1_all.toFloat)))
      .sortBy(_._2, false)
      .collect()
      .foreach(x => redisClient.hset("line_1", s"${x._1}", s"${x._2}"))

    //2号线
    val line_2_enter = lines.filter(line => (line.split(",")(3) == "2")).map(x => (x.split(',')(2), 1)).reduceByKey(_ + _)
    val line_2_out = line_diff.filter(line => (line.split(",")(7) == "2")).map(x => (x.split(',')(6), 1)).reduceByKey(_ + _)
    val line_2 = line_2_enter.join(line_2_out)
      .map(x => (x._1, ((x._2._1 + x._2._2) / line_2_all.toFloat)))
      .sortBy(_._2, false)
      .collect()
      .foreach(x => redisClient.hset("line_2", s"${x._1}", s"${x._2}"))

    //3号线
    val line_3_enter = lines.filter(line => (line.split(",")(3) == "3")).map(x => (x.split(',')(2), 1)).reduceByKey(_ + _)
    val line_3_out = line_diff.filter(line => (line.split(",")(7) == "3")).map(x => (x.split(',')(6), 1)).reduceByKey(_ + _)
    val line_3 = line_3_enter.join(line_3_out)
      .map(x => (x._1, ((x._2._1 + x._2._2) / line_3_all.toFloat)))
      .sortBy(_._2, false)
      .collect()
      .foreach(x => redisClient.hset("line_3", s"${x._1}", s"${x._2}"))

    //4号线
    val line_4_enter = lines.filter(line => (line.split(",")(3) == "4")).map(x => (x.split(',')(2), 1)).reduceByKey(_ + _)
    val line_4_out = line_diff.filter(line => (line.split(",")(7) == "4")).map(x => (x.split(',')(6), 1)).reduceByKey(_ + _)
    val line_4 = line_4_enter.join(line_4_out)
      .map(x => (x._1, ((x._2._1 + x._2._2) / line_4_all.toFloat)))
      .sortBy(_._2, false)
      .collect()
      .foreach(x => redisClient.hset("line_4", s"${x._1}", s"${x._2}"))


    //10.线路收入排行---------柱状图
    val line_enter_money = Station.filter(line => (line.trim().length > 0) && (line.split(",").length == 17))
      .map(x => (x.split(",")(3), x.split(",")(16).toInt))
      .reduceByKey(_ + _)
    val line_out_money = Station.filter(line => (line.trim().length > 0) &&
      (line.split(",").length == 17) &&
      (line.split(",")(3) != line.split(",")(7))) //出战不等于入站
      .map(x => (x.split(",")(7), x.split(",")(16).toInt))
      .reduceByKey(_ + _)
    val line_money_all = line_enter_money.join(line_out_money)
      .map(x => (x._1, x._2._1 + x._2._2))
      .sortBy(_._2, false)
      .collect()
      .foreach(x => redisClient.hset("line_money_all", s"${x._1}", s"${x._2}"))
    //10.各站点对线路收入的贡献---------四个饼---------画在一张图上

    val line_money_all2 = line_enter_money.join(line_out_money)
      .map(x => (x._1, ((x._2._1 + x._2._2))))
      .sortBy(_._2, false)
    val lines_money = line_money_all2.collect().toList

    var line_money_1 = 0
    var line_money_2 = 0
    var line_money_3 = 0
    var line_money_4 = 0

    for (line <- lines_money) {
      if (line._1 == "1") {
        line_money_1 = line._2
      } else if (line._1 == "2") {
        line_money_2 = line._2
      } else if (line._1 == "3") {
        line_money_3 = line._2
      } else if (line._1 == "4") {
        line_money_4 = line._2
      }
    }

    //1号线
    val line_enter_1 = lines.filter(line => (line.split(",")(3) == "1"))
      .map(x => (x.split(",")(2), x.split(",")(16).toInt)).reduceByKey(_ + _)

    val line_out_1 = line_diff.filter(line => (line.split(",")(7) == "1"))
      .map(x => (x.split(",")(6), x.split(",")(16).toInt)).reduceByKey(_ + _)

    val linemoney_1 = line_enter_1.join(line_out_1)
      .map(x => (x._1, ((x._2._1 + x._2._2) / line_money_1.toFloat)))
      .sortBy(_._2, false)
      .collect()
      .foreach(x => redisClient.hset("linemoney_1", s"${x._1}", s"${x._2}"))

    //2号线
    val line_enter_2 = lines.filter(line => (line.split(",")(3) == "2"))
      .map(x => (x.split(",")(2), x.split(",")(16).toInt)).reduceByKey(_ + _)
    val line_out_2 = line_diff.filter(line => (line.split(",")(7) == "2"))
      .map(x => (x.split(",")(6), x.split(",")(16).toInt)).reduceByKey(_ + _)
    val linemoney_2 = line_enter_2.join(line_out_2)
      .map(x => (x._1, ((x._2._1 + x._2._2) / line_money_2.toFloat)))
      .sortBy(_._2, false)
      .collect()
      .foreach(x => redisClient.hset("linemoney_2", s"${x._1}", s"${x._2}"))

    //3号线
    val line_enter_3 = lines.filter(line => (line.split(",")(3) == "3"))
      .map(x => (x.split(",")(2), x.split(",")(16).toInt)).reduceByKey(_ + _)
    val line_out_3 = line_diff.filter(line => (line.split(",")(7) == "3"))
      .map(x => (x.split(",")(6), x.split(",")(16).toInt)).reduceByKey(_ + _)
    val linemoney_3 = line_enter_3.join(line_out_3)
      .map(x => (x._1, ((x._2._1 + x._2._2) / line_money_3.toFloat)))
      .sortBy(_._2, false)
      .collect()
      .foreach(x => redisClient.hset("linemoney_3", s"${x._1}", s"${x._2}"))

    //4号线
    val line_enter_4 = lines.filter(line => (line.split(",")(3) == "4"))
      .map(x => (x.split(",")(2), x.split(",")(16).toInt)).reduceByKey(_ + _)
    val line_out_4 = line_diff.filter(line => (line.split(",")(7) == "4"))
      .map(x => (x.split(",")(6), x.split(",")(16).toInt)).reduceByKey(_ + _)
    val linemoney_4 = line_enter_4.join(line_out_4)
      .map(x => (x._1, ((x._2._1 + x._2._2) / line_money_4.toFloat)))
      .sortBy(_._2, false)
      .collect()
      .foreach(x => redisClient.hset("linemoney_4", s"${x._1}", s"${x._2}"))


    //12.不同消费金额次数排行---------饼图
    val money = Station.filter(line => (line.trim().length > 0) && (line.split(",").length == 17)) //去除空行和属性数不为17的数据
      .map(x => (x.split(",")(16), 1)) //将金额提取出来(“票价”，1)
      .reduceByKey(_ + _) //==>groupbukey().map((_.1, _.2.sum())),按key聚合同一个map
      .sortBy(_._2, false) //按value进行从大到小排序
      .collect()
      .foreach(x => redisClient.hset("money", s"${x._1}", s"${x._2}"))
    //13.不同乘车区间客流量排行--------单程
    val Ride_section = Station.filter(line => (line.trim().length > 0) && (line.split(",").length == 17)) //去除空行和属性数不为17的数据
      .map(x => ((x.split(",")(2), x.split(",")(6)), 1)) //(（出战，入站）,1)
      .reduceByKey(_ + _) //==>groupbukey().map((_.1, _.2.sum())),按key聚合同一个map
      .sortBy(_._2, false) //按value进行从大到小排序
      .collect()
      .foreach(x => redisClient.hset("Ride_section", s"${x._2}", s"${x._1}"))
    //14.不同线路区间客流排行---------单程
    val line_section = Station.filter(line => (line.trim().length > 0) && (line.split(",").length == 17)) //去除空行和属性数不为17的数据
      .map(x => ((x.split(",")(3), x.split(",")(7)), 1)) //将金额提取出来((入站线，出站线)，1)
      .reduceByKey(_ + _) //==>groupbukey().map((_.1, _.2.sum())),按key聚合同一个map
      .sortBy(_._2, false) //按value进行从大到小排序
      .collect()
      .foreach(x => redisClient.hset("line_section", s"${x._1}", s"${x._2}"))
    //15.不同年龄付款方式占比--------饼图-----三个
    val age1 = Station.filter(line => (line.trim().length > 0) && (line.split(",").length == 17)) //去除空行和属性数不为6的数据
      .map(x => {
        val payment = if (x.split(",")(15) == "地铁票")
          "地铁票"
        else if (x.split(",")(15) == "码")
          "码"
        else if (x.split(",")(15) == "一卡通")
          "一卡通"
        val age_section = if (x.split(",")(14) <= "44")
          "青年人"
        else if (x.split(",")(14) <= "59")
          "中年人"
        else if (x.split(",")(14) >= "60")
          "老年人"
        ((age_section, payment), 1)
      })
      .reduceByKey(_ + _) //==>groupbukey().map((_.1, _.2.sum())),按key聚合同一个map
      .sortBy(_._2, false) //按value进行从大到小排序
      .collect()
      .foreach(x => redisClient.hset("age1", s"${x._1}", s"${x._2}"))
    //16.不同节假日对客流量影响
    val holiday = holidayRDD.map(line => (line.split(",")(1), line.split(",")(0)))
    val subway_day = Station.filter(line => (line.trim().length > 0) && (line.split(",").length == 17))
      .map(x => (x.split(",")(4), 1))
    val is_holiday = holiday.join(subway_day)
      .map(x => x._2)
      .reduceByKey(_ + _)
      .sortBy(_._2, false)
      .collect()
      .foreach(x => redisClient.hset("holiday", s"${x._1}", s"${x._2}"))
    //17.不同天气对客流量的影响---------
    val weather = Station.filter(line => (line.trim().length > 0) && (line.split(",").length == 17)) //去除空行和属性数不为17的数据
      .map(x => (x.split(",")(11), 1)) //
      .reduceByKey(_ + _) //==>groupbukey().map((_.1, _.2.sum())),按key聚合同一个map
      .sortBy(_._2, false) //按value进行从大到小排序
      .collect()
      .foreach(x => redisClient.hset("weather", s"${x._1}", s"${x._2}"))

    spark.stop()
  }
}
